/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.processors.huawei.obs;

import com.obs.services.ObsClient;
import com.obs.services.exception.ObsException;
import com.obs.services.model.*;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processors.huawei.credentials.provider.service.HuaweiCredentialsProviderControllerService;
import org.apache.nifi.processors.huawei.obs.encryption.StandardOBSEncryptionService;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import static org.apache.nifi.processors.huawei.common.PropertyDescriptors.*;
import static org.junit.Assert.fail;

/**
 * Base class for OBS Integration Tests. Establishes a bucket and helper methods for creating test scenarios
 */
public abstract class AbstractOBSIT {

    protected final static String ak = "your ak";
    protected final static String sk = "your sk";
    protected final static String SAMPLE_FILE_RESOURCE_NAME = "/hello.txt";
    protected final static String REGION_NAME = "cn-south-1";

    // the endpoint of REGION_NAME
    protected final static String ENDPOINT = "obs." + REGION_NAME +".myhuaweicloud.com";

    // Static so multiple Tests can use same client, you may need to clean up uploaded files at the end of the method
    protected static ObsClient client;

    protected static final String bucketName = "nifi-test" + System.currentTimeMillis() + "-" + REGION_NAME;
    @BeforeAll
    protected static void createClientAndBucket() {
        // Creates a client and bucket for this test
        try {
            client = new ObsClient(ak, sk, OBSRegions.fromName(REGION_NAME).getEndpoint());
            if (client.headBucket(bucketName)) {
                fail("Bucket " + bucketName + " exists. Choose a different bucket name to continue test");
            }
            CreateBucketRequest request = new CreateBucketRequest(bucketName, REGION_NAME);
            client.createBucket(request);
            if (!client.headBucket(bucketName)) {
                fail("Setup incomplete, tests will fail");
            }
        } catch (final ObsException e) {
            e.printStackTrace();
            fail("Can't create the key " + bucketName + ": " + e.getLocalizedMessage());
        }
    }

    @AfterAll
    public static void clear() {
        // Empty the bucket before deleting it.
        if (client == null) {
            return;
        }
        try {
            clearObjects();
            client.deleteBucket(bucketName);
        } catch (final ObsException e) {
            System.err.println("Unable to delete bucket " + bucketName + e);
        }
        if (client.headBucket(bucketName)) {
            fail("Incomplete teardown, subsequent tests might fail");
        }
    }

    protected static void clearObjects() {
        if (client == null) {
            return;
        }
        ListVersionsResult objectListing = client.listVersions(bucketName);
        while (true) {
            for (VersionOrDeleteMarker versionOrDeleteMarker : objectListing.getVersions()) {
                client.deleteObject(bucketName, versionOrDeleteMarker.getObjectKey(), versionOrDeleteMarker.getVersionId());
            }
            if (!objectListing.isTruncated()) {
                break;
            }
        }
    }

    protected static void clearObjects(String... keys) {
        if (client == null || ArrayUtils.isEmpty(keys)) {
            return;
        }
        List<String> list = Arrays.asList(keys);
        ObjectListing objectListing = client.listObjects(bucketName);
        while (true) {
            for (ObsObject objectSummary : objectListing.getObjects()) {
                if (!list.contains(objectSummary.getObjectKey())) {
                    continue;
                }
                client.deleteObject(bucketName, objectSummary.getObjectKey());
            }
            if (!objectListing.isTruncated()) {
                break;
            }
        }
    }

    protected synchronized void sleep() throws InterruptedException {
        Thread.sleep(1000);
    }

    protected void putTestFile(String... keys) throws ObsException, InterruptedException {
        if (ArrayUtils.isEmpty(keys)) {
            return;
        }
        File file = getFileFromResourceName(SAMPLE_FILE_RESOURCE_NAME);

        for (String key : keys) {
            PutObjectRequest putRequest = new PutObjectRequest(bucketName,  key, file);
            client.putObject(putRequest);
            Thread.sleep(10);
        }
    }

    protected void putTestFileEncrypted(String key, File file) throws ObsException, FileNotFoundException {
        PutObjectRequest putRequest = new PutObjectRequest(bucketName, key, new FileInputStream(file));
        putRequest.setSseKmsHeader(new SseKmsHeader());
        client.putObject(putRequest);
    }

    protected void putFileWithUserMetadata(String key, File file, Map<String, Object> userMetadata) throws ObsException, FileNotFoundException {
        ObjectMetadata objectMetadata = new ObjectMetadata();
        objectMetadata.setUserMetadata(userMetadata);
        PutObjectRequest putRequest = new PutObjectRequest(bucketName, key, new FileInputStream(file));
        putRequest.setMetadata(objectMetadata);
        client.putObject(putRequest);
    }

    protected Path getResourcePath(String resourceName) {
        Path path = null;

        try {
            path = Paths.get(getClass().getResource(resourceName).toURI());
        } catch (URISyntaxException e) {
            fail("Resource: " + resourceName + " does not exist" + e.getLocalizedMessage());
        }

        return path;
    }

    protected File getFileFromResourceName(String resourceName) {
        URI uri = null;
        try {
            uri = this.getClass().getResource(resourceName).toURI();
        } catch (URISyntaxException e) {
            fail("Cannot proceed without File : " + resourceName);
        }

        return new File(uri);
    }

    protected static void addEncryptionService(TestRunner runner, AllowableValue allowableValue, String keyValue) {
        ControllerService controllerService = new StandardOBSEncryptionService();
        try {
            runner.addControllerService("standardObsEncryptionService", controllerService);
        } catch (InitializationException e) {
            throw new RuntimeException(e);
        }
        runner.setProperty(controllerService, StandardOBSEncryptionService.ENCRYPTION_STRATEGY, allowableValue);
        runner.setProperty(controllerService, StandardOBSEncryptionService.ENCRYPTION_VALUE, keyValue);
        runner.setProperty(controllerService, StandardOBSEncryptionService.KMS_REGION, REGION_NAME);
        runner.enableControllerService(controllerService);
        runner.assertValid(controllerService);
        runner.setProperty(ENCRYPTION_SERVICE, "standardObsEncryptionService");
    }

    protected static TestRunner initRunner(Processor processor) {
        final TestRunner runner = TestRunners.newTestRunner(processor);
        runner.setProperty(OBS_REGION, REGION_NAME);
        runner.setProperty(BUCKET, bucketName);
        runner.setProperty(ACCESS_KEY, ak);
        runner.setProperty(SECRET_KEY, sk);
        return runner;
    }

    protected static TestRunner initRunner(Processor processor, String endpoint) {
        final TestRunner runner = TestRunners.newTestRunner(processor);
        runner.setProperty(OBS_REGION, "NotFound");
        runner.setProperty(ENDPOINT_OVERRIDE_URL, endpoint);
        runner.setProperty(BUCKET, bucketName);
        runner.setProperty(ACCESS_KEY, ak);
        runner.setProperty(SECRET_KEY, sk);
        return runner;
    }

    protected static TestRunner initCredentialsRunner(Processor processor) throws InitializationException {
        final TestRunner runner = TestRunners.newTestRunner(processor);
        runner.setProperty(OBS_REGION, REGION_NAME);
        runner.setProperty(BUCKET, bucketName);
        final HuaweiCredentialsProviderControllerService serviceImpl = new HuaweiCredentialsProviderControllerService();
        runner.addControllerService("huaweiCredentialsProvider", serviceImpl);
        runner.setProperty(serviceImpl, ACCESS_KEY, ak);
        runner.setProperty(serviceImpl, SECRET_KEY, sk);
        runner.enableControllerService(serviceImpl);
        runner.assertValid(serviceImpl);
        runner.setProperty(HUAWEI_CREDENTIALS_PROVIDER_SERVICE, "huaweiCredentialsProvider");
        return runner;
    }
}